/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
 * Description: pingmesh 接口实现
 */
 
#include <thread>
#include <chrono>
#include "securec.h"
#include "adapter_hccp.h"
#include "adapter_tdt.h"
#include "adapter_hal.h"
#include "adapter_rts.h"
#include "adapter_rts_common.h"
#include "dispatcher_task_types.h"
#include "network_manager_pub.h"
#include "externalinput.h"
#include "dlra_function.h"
#include "sal_pub.h"
#include "ping_mesh.h"

namespace hccl {
constexpr int HOP_MAX_TIMES = 64; // 最大跳数
constexpr int LOG_CHK_TIMES = 50; // 需要打印日志的轮询次数
constexpr u32 WR_DEPTH_MULTIPLE = 4; // wr深度扩展倍数
constexpr u32 BYTE_PER_TARGET_DEFAULT = 2048; // 记录每个target的result时默认需要的buffersize大小
constexpr u32 BYTE_IPV4_SHIFT_IN_GID = 12; // ipv4地址相比于gid首地址的偏移值
constexpr u32 RPING_PAYLOAD_REFILL_LEN = 136; // payload头需要重填的长度
constexpr u32 RPING_PAYLOAD_RSVD_LEN = 44;    // payload头需要清零的长度
 
PingMesh::PingMesh()
{}
 
PingMesh::~PingMesh()
{
    if (!isDeinited_) { // 如果没有主动释放过资源，析构时需要释放一下
        HccnRpingDeinit(deviceLogicId_);
    }
}
 
inline HcclResult UninitStateCheck(RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(nextState) {
        case RpingState::UNINIT:
            HCCL_INFO("[HCCN][StateCheck]Device is uninited, does not need uninit.");
            break;
        case RpingState::INITED:
            break;
        case RpingState::READY:
            HCCL_ERROR("[HCCN][StateCheck]Device is not inited yet.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        case RpingState::RUN:
            HCCL_ERROR("[HCCN][StateCheck]Device is not inited yet.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        case RpingState::STOP:
            HCCL_ERROR("[HCCN][StateCheck]Device is not inited yet.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Undefinied behavior.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline HcclResult InitedStateCheck(RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(nextState) {
        case RpingState::UNINIT:
            break;
        case RpingState::INITED:
            HCCL_INFO("[HCCN][StateCheck]Device is inited already.");
            break;
        case RpingState::READY:
            break;
        case RpingState::RUN:
            HCCL_ERROR("[HCCN][StateCheck]Device is not ready.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        case RpingState::STOP:
            HCCL_ERROR("[HCCN][StateCheck]Device is not ready.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Undefinied behavior.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline HcclResult ReadyStateCheck(RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(nextState) {
        case RpingState::UNINIT:
            break;
        case RpingState::INITED:
            break;
        case RpingState::READY:
            break;
        case RpingState::RUN:
            break;
        case RpingState::STOP:
            HCCL_ERROR("[HCCN][StateCheck]Device has not run tasks yet.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Undefinied behavior.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline HcclResult RunStateCheck(RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(nextState) {
        case RpingState::UNINIT:
            HCCL_WARNING("[HCCN][StateCheck]Make sure the task is finished and result has already gotten.");
            break;
        case RpingState::INITED:
            break;
        case RpingState::READY:
            break;
        case RpingState::RUN:
            break;
        case RpingState::STOP:
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Undefinied behavior.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline HcclResult StopStateCheck(RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(nextState) {
        case RpingState::UNINIT:
            break;
        case RpingState::INITED:
            break;
        case RpingState::READY:
            break;
        case RpingState::RUN:
            break;
        case RpingState::STOP:
            HCCL_WARNING("[HCCN][StateCheck]Task is stopped.");
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Undefinied behavior.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline HcclResult RpingstateCheck(RpingState currState, RpingState nextState)
{
    HcclResult ret = HCCL_SUCCESS;
    switch(currState) {
        case RpingState::UNINIT:
            ret = UninitStateCheck(nextState);
            break;
        case RpingState::INITED:
            ret = InitedStateCheck(nextState);
            break;
        case RpingState::READY:
            ret = ReadyStateCheck(nextState);
            break;
        case RpingState::RUN:
            ret = RunStateCheck(nextState);
            break;
        case RpingState::STOP:
            ret = StopStateCheck(nextState);
            break;
        default:
            HCCL_ERROR("[HCCN][StateCheck]Current state doesn't exist.");
            ret = HCCL_E_NOT_SUPPORT;
            break;
    }
 
    return ret;
}
 
inline void TsdProcessOpenInit(ProcOpenArgs &openArgs, ProcExtParam *extParam)
{
    int subPid = 0;
    std::string extPam[TSD_EXT_PARA_NUM] = {std::string("--hdcType=" + std::to_string(HDC_SERVICE_TYPE_RDMA_V2)),
                                            std::string("--whiteListStatus=" + std::to_string(WHITE_LIST_CLOSE))};
    for (u32 i = 0; i < TSD_EXT_PARA_NUM; i++) {
        extParam[i].paramInfo = extPam[i].c_str();
        extParam[i].paramLen = extPam[i].size();
    }
    openArgs.procType = TSD_SUB_PROC_HCCP;
    openArgs.filePath = nullptr;
    openArgs.extParamList = extParam;
    openArgs.extParamCnt = TSD_EXT_PARA_NUM;
    openArgs.pathLen = 0UL;
    openArgs.subPid = &subPid;  // 进程号, 填充后返回, 给内存地址即可
    HCCL_INFO("[HCCN]PingInitAttrInit success. ParaNum[%llu] extPar0[%s] size[%lu], extPar1[%s] size[%lu]",
               TSD_EXT_PARA_NUM, extParam[0].paramInfo, extParam[0].paramLen, extParam[1].paramInfo, extParam[1].paramLen);
}
 
inline void RpingInitAttrInit(u32 deviceId, HcclIpAddress ipAddr, u32 port, u32 nodeNum, u32 bufferSize, u32 sl, u32 tc,
                              ping_init_attr &initAttr)
{
    u32 maxWrDepth = nodeNum * WR_DEPTH_MULTIPLE;
    maxWrDepth = (maxWrDepth > DEFAULT_OPBASE_MAX_SEND_WR) ? DEFAULT_OPBASE_MAX_SEND_WR : maxWrDepth;
    
    initAttr.version = 0; // 暂时无用，默认给0
    initAttr.mode = NETWORK_OFFLINE; // net work mode 枚举值
    initAttr.rdev_info.phy_id = deviceId;
    initAttr.rdev_info.family = ipAddr.GetFamily(); // AF_INET(ipv4) or AF_INET6(ipv6)
    initAttr.rdev_info.local_ip.addr = ipAddr.GetBinaryAddress().addr;
    initAttr.rdev_info.local_ip.addr6 = ipAddr.GetBinaryAddress().addr6;
    initAttr.buffer_size = bufferSize == 0 ? (maxWrDepth * BYTE_PER_TARGET_DEFAULT) : bufferSize; // 发送接收缓存区大小
 
    // client的初始化信息
    initAttr.client.cq_attr.send_cq_depth = maxWrDepth;
    initAttr.client.cq_attr.recv_cq_depth = maxWrDepth;
    initAttr.client.cq_attr.send_cq_comp_vector = 0; // 一组cqe组成的集合，这里给0
    initAttr.client.cq_attr.recv_cq_comp_vector = 1; // 一组cqe组成的集合，这里给1
    initAttr.client.qp_attr.qp_cap.max_send_wr = maxWrDepth;
    initAttr.client.qp_attr.qp_cap.max_recv_wr = maxWrDepth;
    initAttr.client.qp_attr.qp_cap.max_send_sge = DEFAULT_MAX_SEND_SGE;
    initAttr.client.qp_attr.qp_cap.max_recv_sge = DEFAULT_MAX_RECV_SGE;
    initAttr.client.qp_attr.qp_cap.max_inline_data = DEFAULT_MAX_INLINE_DATA;
    initAttr.client.qp_attr.udp_sport = 0;
 
    // server的初始化信息
    initAttr.server.cq_attr.send_cq_depth = maxWrDepth;
    initAttr.server.cq_attr.recv_cq_depth = maxWrDepth;
    initAttr.server.cq_attr.send_cq_comp_vector = 0; // 一组cqe组成的集合，这里给0
    initAttr.server.cq_attr.recv_cq_comp_vector = 1; // 一组cqe组成的集合，这里给1
    initAttr.server.qp_attr.qp_cap.max_send_wr = maxWrDepth;
    initAttr.server.qp_attr.qp_cap.max_recv_wr = maxWrDepth;
    initAttr.server.qp_attr.qp_cap.max_send_sge = DEFAULT_MAX_SEND_SGE;
    initAttr.server.qp_attr.qp_cap.max_recv_sge = DEFAULT_MAX_RECV_SGE;
    initAttr.server.qp_attr.qp_cap.max_inline_data = DEFAULT_MAX_INLINE_DATA;
    initAttr.server.qp_attr.udp_sport = 0;
 
    // ip协议信息
    initAttr.comm_info.version = 0;
    initAttr.comm_info.flow_label = 0;
    initAttr.comm_info.hop_limit = HOP_MAX_TIMES;
    initAttr.comm_info.qos_attr.sl = sl;
    initAttr.comm_info.qos_attr.tc = tc;
}

inline HcclResult RpingTargetAttrInit(ping_target_info &target, RpingInput input, ping_rdma_info *rdmainfo, bool isAddTarget)
{
    target.remote_info.rdma.version = rdmainfo->version;
    target.remote_info.rdma.gid = rdmainfo->gid;
    target.remote_info.rdma.qpn = rdmainfo->qpn;
    target.remote_info.rdma.qkey = rdmainfo->qkey;
    target.remote_info.ip.addr = input.dip.GetBinaryAddress().addr;
    target.remote_info.ip.addr6 = input.dip.GetBinaryAddress().addr6;
    target.local_info.qos_attr.tc = input.tc;
    target.local_info.qos_attr.sl = input.sl;
    target.local_info.flow_label = 0;
    target.local_info.hop_limit = HOP_MAX_TIMES;
    target.local_info.udp_sport = input.srcPort;
    if (!isAddTarget) { // 并非添加target的时候调用，不需要拷贝payload信息
        return HCCL_SUCCESS;
    }
    if (input.len > PING_USER_PAYLOAD_MAX_SIZE) {
        HCCL_WARNING(
            "[HCCN][HccnRpingAddTarget]Payload length should be less than %u byte.", PING_USER_PAYLOAD_MAX_SIZE + 1);
        target.payload.size = 0;
        return HCCL_SUCCESS;
    }
    target.payload.size = input.len;
    errno_t memRet = memcpy_s(target.payload.buffer, target.payload.size, input.payload, input.len);
    if (memRet != EOK) {
        HCCL_ERROR("[HCCN][HccnRpingAddTarget] Memcpy ret %d, dst:%u, dstMax:%u, src:%u, length:%u",
            memRet, target.payload.buffer, target.payload.size, input.payload, input.len);
        return HCCL_E_MEMORY;
    }

    return HCCL_SUCCESS;
}

inline void RpingResultInfoInit(ping_target_result *resultInfo, std::map<std::string, ping_rdma_info> rdmaInfoMaps,
    RpingInput *input, u32 targetNum)
{
    for (u32 i = 0; i < targetNum; i++) {
        if (rdmaInfoMaps.find(std::string(input[i].dip.GetReadableIP())) == rdmaInfoMaps.end()) {
            HCCL_WARNING("[HCCN][HccnRpingGetResult]Target[%s] info doesn't exist.", input[i].dip.GetReadableIP());
            continue;
        }
        ping_rdma_info *rdmainfo = &rdmaInfoMaps[std::string(input[i].dip.GetReadableIP())];
        resultInfo[i].remote_info.ip.addr = input[i].dip.GetBinaryAddress().addr;
        resultInfo[i].remote_info.ip.addr6 = input[i].dip.GetBinaryAddress().addr6;
        resultInfo[i].remote_info.rdma.version = 0;
        resultInfo[i].remote_info.rdma.gid = rdmainfo->gid;
        resultInfo[i].remote_info.rdma.qpn = rdmainfo->qpn;
        resultInfo[i].remote_info.rdma.qkey = rdmainfo->qkey;

        HCCL_INFO("[HCCN][HccnRpingGetResult]Target[%s] info init success.", input[i].dip.GetReadableIP());
    }
}

inline void GetResultFromReturnValue(ping_target_result *resultInfo, RpingOutput *output, u32 targetNum)
{
    for (u32 i = 0; i < targetNum; i++) {
        output[i].state = resultInfo[i].result.state;
        output[i].txPkt = resultInfo[i].result.summary.send_cnt;
        output[i].rxPkt = resultInfo[i].result.summary.recv_cnt;
        output[i].minRTT = resultInfo[i].result.summary.rtt_min;
        output[i].maxRTT = resultInfo[i].result.summary.rtt_max;
        output[i].avgRTT = resultInfo[i].result.summary.rtt_avg;
    }
}

inline void LogRecordbyTimes(int &count)
{
    // 日志过滤, 50次才打印一次
    if (count % LOG_CHK_TIMES == 0) {
        HCCL_DEBUG("[HCCN][HccnRpingInit]socket is connecting...");
    }
    count++;
}

inline void RemoveMapInfo(RpingInput *input, u32 targetNum, 
                          std::map<std::string, std::shared_ptr<HcclSocket>> &socketMaps,
                          std::map<std::string, ping_rdma_info> &rdmaInfoMaps)
{
    for (u32 i = 0; i < targetNum; i++) {
        socketMaps.erase(std::string(input[i].dip.GetReadableIP()));
        rdmaInfoMaps.erase(std::string(input[i].dip.GetReadableIP()));
    }
}

HcclResult PingMesh::RpingSendInitInfo(u32 deviceId, u32 port, HcclIpAddress ipAddr, ping_init_info initInfo,
                                       std::shared_ptr<HcclSocket> socket)
{
    // 给当前线程添加名字
    SetThreadName("Hccl_PingMesh");

    // 等待client端发送的建链请求
    HcclIpAddress remoteIp = HcclIpAddress();
    std::string tag = "PingMesh" + std::string(ipAddr.GetReadableIP());
    HCCL_INFO("[HCCN][HccnRpingInit]socket tag[%s].", tag.c_str());
    // 持续在后台等待建链，保证可以处理多个client端的建链请求
    while (true) {
        HcclSocket realSocket(tag, netCtx_, remoteIp, 0, HcclSocketRole::SOCKET_ROLE_SERVER);
        CHK_RET(realSocket.Init());
        int count = 0; // 轮询计数
        while (true) {
            bool isStop = connThreadStop_.load();
            if (isStop == true) {
                SaluSleep(ONE_MILLISECOND_OF_USLEEP);
                HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] stop waiting connect.", deviceId);
                break;
            }
            HcclSocketStatus status = realSocket.GetStatus();
            if (status == HcclSocketStatus::SOCKET_OK) {
                HCCL_DEBUG("[HCCN][HccnRpingInit]socket is established. localIp[%s], remoteIp[%s]",
                    realSocket.GetLocalIp().GetReadableIP(), realSocket.GetRemoteIp().GetReadableIP());
                break;
            } else if (status == HcclSocketStatus::SOCKET_CONNECTING) {
                SaluSleep(ONE_MILLISECOND_OF_USLEEP);
                LogRecordbyTimes(count);
                continue;
            } else if (status == HcclSocketStatus::SOCKET_TIMEOUT) {
                HCCL_WARNING("[HCCN][HccnRpingInit]socket connect timeout.");
                break;
            } else {
                HCCL_WARNING("[HCCN][HccnRpingInit]socket connect failed.");
                break;
            }
        }

        // 判断否需要中止线程
        bool isStop = connThreadStop_.load();
        if (isStop == true) {
            HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] background thread stopped.", deviceId);
            break;
        }
    
        // 建链成功，发送rping初始化信息
        u64 sendSize = sizeof(initInfo);
        CHK_RET(realSocket.Send(&initInfo, sendSize));
        HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] rdma info send success.", deviceId);
    }

    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::RpingRecvTargetInfo(void *clientNetCtx, u32 port, HcclIpAddress ipAddr, ping_init_info &recvInfo)
{
    // 确认是否添加过该IP
    if (socketMaps_.find(std::string(ipAddr.GetReadableIP())) != socketMaps_.end()) {
        HCCL_WARNING("[HCCN][HccnRpingAddTarget]IP address[%s] has already exist.", ipAddr.GetReadableIP());
        return HCCL_SUCCESS;
    }
    // socket建链 这里的建链流程与init里的侦听动作应当使用同一套接口
    std::string tag = "PingMesh" + std::string(ipAddr.GetReadableIP());
    HCCL_INFO("[HCCN][HccnRpingAddTarget]socket tag[%s].", tag.c_str());
    std::shared_ptr<HcclSocket> socket = nullptr;
    EXECEPTION_CATCH(
        (socket = std::make_shared<HcclSocket>(tag, clientNetCtx, ipAddr, port, HcclSocketRole::SOCKET_ROLE_CLIENT)),
        return HCCL_E_PTR);
    CHK_SMART_PTR_NULL(socket);
    CHK_RET(socket->Init());
    CHK_RET(socket->Connect());
    auto startTime = std::chrono::steady_clock::now();
    auto timeout = std::chrono::seconds(120);
    while (true) {
        if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
            HCCL_ERROR("[HCCN][HccnRpingAddTarget] Get socket timeout! timeout [%ld s]", timeout);
            return HCCL_E_TIMEOUT;
        }
 
        auto status = socket->GetStatus();
        if (status == HcclSocketStatus::SOCKET_CONNECTING) {
            SaluSleep(ONE_MILLISEC);
            HCCL_INFO("[HCCN][HccnRpingAddTarget] connecting to server [%s] port [%u]", ipAddr.GetReadableIP(), port);
            continue;
        } else if (status != HcclSocketStatus::SOCKET_OK) {
            HCCL_ERROR("[HCCN][HccnRpingAddTarget] Get socket failed, ret [%d]", status);
            return HCCL_E_TCP_CONNECT;
        } else {
            HCCL_INFO("[HCCN][HccnRpingAddTarget] Get socket success with server [%s] port [%u]",
                ipAddr.GetReadableIP(), port);
            break;
        }
    }
 
    // 接收发送的信息
    u32 recvBufLen = sizeof(recvInfo);
    CHK_RET(socket->Recv(&recvInfo, recvBufLen));
    HCCL_INFO("[HCCN][HccnRpingAddTarget]Server[%s] info received success.", ipAddr.GetReadableIP());
 
    // 记录socket
    socketMaps_.insert({std::string(ipAddr.GetReadableIP()), socket});
 
    return HCCL_SUCCESS;
}

inline RpingLinkState ConvertHcclSocketStatus(HcclSocketStatus socketStatus)
{
    RpingLinkState status = RpingLinkState::DISCONNECTED;
    switch (socketStatus) {
        case HcclSocketStatus::SOCKET_INIT:
            status = RpingLinkState::DISCONNECTED;
            break;
        case HcclSocketStatus::SOCKET_OK:
            status = RpingLinkState::CONNECTED;
            break;
        case HcclSocketStatus::SOCKET_TIMEOUT:
            status = RpingLinkState::TIMEOUT;
            break;
        case HcclSocketStatus::SOCKET_CONNECTING:
            status = RpingLinkState::CONNECTING;
            break;
        default:
            status = RpingLinkState::ERROR;
            break;
    }
    return status;
}

HcclResult PingMesh::HccnRpingInit(u32 deviceId, u32 mode, HcclIpAddress ipAddr, u32 port, u32 nodeNum, u32 bufferSize,
    u32 sl, u32 tc)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::INITED));
 
    // 当前只支持RoCE
    LinkType netMode = static_cast<LinkType>(mode);
    if (netMode != LinkType::LINK_ROCE) {
        HCCL_ERROR("[HCCN][HccnRpingInit]only support ROCE mode.");
        return HCCL_E_NOT_SUPPORT;
    }
 
    // 获取物理id
    deviceLogicId_ = deviceId;
    CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
    if (deviceId != static_cast<u32>(deviceLogicId_)) {
        HCCL_ERROR("[HCCN][HccnRpingInit]Input device logicId[%u] don't match real logicId[%s].", deviceId, deviceLogicId_);
        return HCCL_E_PARA;
    }
    HCCL_INFO("[HCCN][HccnRpingInit]Device logic id is [%d], phy id is [%u].", deviceLogicId_, devicePhyId_);
 
    // 拉起hccp进程
    ProcExtParam extParam[TSD_EXT_PARA_NUM] {};
    TsdProcessOpenInit(hccpProcessInfo_, extParam);
    CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionHeterogInit());
    CHK_RET(hrtTsdProcessOpen(deviceId, &hccpProcessInfo_));
    HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] open process success, subPid[%d].", deviceId, *(hccpProcessInfo_.subPid));
 
    // hccp侧初始化ping mesh资源
    ra_init_config config = { devicePhyId_, static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_DEVICE),
        HDC_SERVICE_TYPE_RDMA_V2 };
    u32 rpingInterfaceVersion = 0;
    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).PingMeshRaPingInit(deviceLogicId_, devicePhyId_, &config));
    CHK_RET(hrtRaGetInterfaceVersion(devicePhyId_, RPING_INTERFACE_OPCODE, &rpingInterfaceVersion));
    if (rpingInterfaceVersion < RPING_INTERFACE_VERSION) {
        HCCL_WARNING("[HCCN][HccnRpingInit]this package[%u] does not support rpingInterface for device.",
            rpingInterfaceVersion);
        return HCCL_SUCCESS;
    }
    HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] init hccp success.", deviceId);
 
    ping_init_attr initAttr{};
    void *pingHandle = nullptr;
    RpingInitAttrInit(deviceId, ipAddr, port, nodeNum, bufferSize, sl, tc, initAttr);
    CHK_RET(hrtRaPingInit(&initAttr, &initInfo_, &pingHandle));
    HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] init success.", deviceId);
 
    // 建链并发送初始化信息
    CHK_RET(HcclNetOpenDev(&netCtx_, NicType::DEVICE_NIC_TYPE, devicePhyId_, deviceLogicId_, ipAddr));
    CHK_PTR_NULL(netCtx_);
    socket_ = std::make_shared<HcclSocket>(netCtx_, port);
    // 初始化socket并启动侦听
    CHK_RET(socket_->Init());
    CHK_RET(SetTcpMode(true));
    CHK_RET(socket_->Listen());
    HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] starts listen port[%u].", deviceId, port);
    // 等待客户端建链
    connThread_.reset(new (std::nothrow)
                          std::thread(&PingMesh::RpingSendInitInfo, this, deviceId, port, ipAddr, initInfo_, socket_));
    CHK_SMART_PTR_NULL(connThread_);
 
    // 绑定信息
    pingHandle_ = pingHandle;
    rpingState_ = RpingState::INITED;
    ipAddr_ = &ipAddr;
    isUsePayload_ = bufferSize == 0 ? false : true;
 
    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingDeinit(u32 deviceId)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::UNINIT));
    CHK_PRT_RET(rpingState_ == RpingState::UNINIT,
        HCCL_WARNING("[HCCN][HccnRpingDeinit]Device[%u] has not inited.", deviceId), HCCL_SUCCESS);

    // 释放payload内存
    if (payload_ != nullptr) {
        delete[] payload_;
        payload_ = nullptr;
    }

    // 手动结束背景线程
    connThreadStop_.store(true);
    if (connThread_->joinable()) {
        connThread_->join();
        HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] end background thread success.", deviceId);
    }

    // 清空map
    for (auto &socket: socketMaps_) {
        CHK_RET(socket.second->DeInit());
    }
    socketMaps_.clear();
    HCCL_INFO("[HCCN][HccnRpingDeinit]Socket map clear.");
    rdmaInfoMaps_.clear();
    HCCL_INFO("[HCCN][HccnRpingDeinit]Rdma info map clear.");

    // 关闭socket链路
    if ((socket_ != nullptr) && (!isSocketClosed_)) {
        CHK_RET(socket_->DeInit());
        isSocketClosed_ = true;
        HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] deinit socket success.", deviceId);
    }

    // 释放资源
    if (netCtx_ != nullptr) {
        HcclNetCloseDev(netCtx_);
        netCtx_ = nullptr;
    }

    if (pingHandle_ == nullptr) {
        HCCL_WARNING("[HCCN][HccnRpingDeinit]Device[%u] don't need to deinit because it is not inited.", deviceId);
        return HCCL_SUCCESS;
    }
    CHK_RET(hrtRaPingDeinit(pingHandle_));
    HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] deinit hccp success.", deviceId);

    // 关闭hccp进程
    CHK_RET(NetworkManager::GetInstance(static_cast<s32>(deviceId)).PingMeshRaPingDeinit());
    if (hccpProcessInfo_.subPid != nullptr) {
        ProcStatusParam closeInfo{};
        closeInfo.pid = *(hccpProcessInfo_.subPid);
        closeInfo.procType = hccpProcessInfo_.procType;
        CHK_RET(hrtProcessCloseSubProcList(deviceId, &closeInfo, 1UL));
        HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] close hccp process success.", deviceId);
    }
    isDeinited_ = true;
    rpingState_ = RpingState::UNINIT;
    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingAddTarget(u32 deviceId, u32 targetNum, RpingInput *input)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::READY));
 
    // 调用hccp接口添加目标
    if (pingHandle_ == nullptr) {
        HCCL_ERROR("[HCCN][HccnRpingAddTarget]Device[%u] cannot add targets because it is not inited.", deviceId);
        return HCCL_E_NOT_FOUND;
    }

    HcclResult ret = HCCL_SUCCESS;
    ping_target_info target[1] = { {0} }; // hccp侧只能一个一个处理，因此数组大小固定为1
    for (u32 i = 0; i < targetNum; i++) {
        ping_init_info recvInfo;
        ret = RpingRecvTargetInfo(netCtx_, input[i].port, input[i].dip, recvInfo);
        if (ret != HCCL_SUCCESS) {
            HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] added failed because of error[%d].",
                input[i].dip.GetReadableIP(), ret);
            break;
        }
        ping_rdma_info *rdmainfo = &(recvInfo.client);
        if (rdmaInfoMaps_.find(std::string(input[i].dip.GetReadableIP())) != rdmaInfoMaps_.end()) {
            HCCL_WARNING("[HCCN][HccnRpingAddTarget]Target[%s] has already added.", input[i].dip.GetReadableIP());
            continue;
        }
        rdmaInfoMaps_.insert(std::pair<std::string, ping_rdma_info>(std::string(input[i].dip.GetReadableIP()), recvInfo.client));

        if (payloadLenMap_.find(std::string(input[i].dip.GetReadableIP())) != payloadLenMap_.end()) {
            HCCL_RUN_INFO("[HCCN][HccnRpingAddTarget]Target[%s] has already added.", input[i].dip.GetReadableIP());
            continue;
        }
        payloadLenMap_.insert(std::pair<std::string, u32>(std::string(input[i].dip.GetReadableIP()), input[i].len));

        ret = RpingTargetAttrInit(target[0], input[i], rdmainfo, true);
        if (ret != HCCL_SUCCESS) {
            HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] payload added failed.", input[i].dip.GetReadableIP());
            break;
        }

        ret = hrtRaPingTargetAdd(pingHandle_, target, 1); // hccp侧只能一个一个处理，因此数组大小固定为1
        if (ret != HCCL_SUCCESS) {
            HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] added failed because of error[%d]",
                input[i].dip.GetReadableIP(), ret);
            break;
        }
        HCCL_INFO("[HCCN][HccnRpingAddTarget]Target[%s] added success.", input[i].dip.GetReadableIP());
        rpingTargetNum_++;
    }
 
    if ((ret == HCCL_SUCCESS) && (rpingState_ == RpingState::INITED)) { // 从初始化完成的状态切换到ready to start的状态
        rpingState_ = RpingState::READY;
    }
 
    return ret;
}
 
HcclResult PingMesh::HccnRpingRemoveTarget(u32 deviceId, u32 targetNum, RpingInput *input)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::READY));
    CHK_RET(RpingstateCheck(rpingState_, RpingState::INITED)); // 所有目标都被移除时回到READY前的状态
 
    if (pingHandle_ == nullptr) {
        HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Device[%u] cannot add targets because it is not inited.", deviceId);
        return HCCL_E_NOT_FOUND;
    }
 
    // 调用hccp接口删除目标
    HcclResult ret = HCCL_SUCCESS;
    ping_target_comm_info *target = new (std::nothrow) ping_target_comm_info[targetNum];
    std::shared_ptr<HcclSocket> socket = nullptr;
    for (u32 i = 0; i < targetNum; i++) {
        // 删除链路
        if (socketMaps_.find(std::string(input[i].dip.GetReadableIP())) == socketMaps_.end()) {
            HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Socket[%s] doesn't exist.", input[i].dip.GetReadableIP());
            ret = HCCL_E_NOT_FOUND;
            break;
        }
        socket = socketMaps_[std::string(input[i].dip.GetReadableIP())];
        ret = socket->DeInit();
        if (ret != HCCL_SUCCESS) {
            HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Socket[%s] deinit failed.", input[i].dip.GetReadableIP());
            break;
        }
        if (rdmaInfoMaps_.find(std::string(input[i].dip.GetReadableIP())) == rdmaInfoMaps_.end()) {
            HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Target[%s] doesn't exist.", input[i].dip.GetReadableIP());
            ret = HCCL_E_NOT_FOUND;
            break;
        }
        ping_rdma_info *rdmainfo = &rdmaInfoMaps_[std::string(input[i].dip.GetReadableIP())];
        ping_target_info targetInfo { 0 };
        RpingTargetAttrInit(targetInfo, input[i], rdmainfo, false);
        target[i] = targetInfo.remote_info;
        rpingTargetNum_--;
    }
    if (ret != HCCL_SUCCESS) {
        delete[] target;
        HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Target info is not correct.");
        return ret;
    }
    ret = hrtRaPingTargetDel(pingHandle_, target, targetNum);
    delete[] target;
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Device[%u] remove target failed.", deviceId);
        return ret;
    }
    HCCL_INFO("[HCCN][HccnRpingRemoveTarget]Device[%u] remove targets success.", deviceId);
    
    // 清除需要删掉的socket和rdma信息
    RemoveMapInfo(input, targetNum, socketMaps_, rdmaInfoMaps_);
    if (rpingTargetNum_ <= 0) { // 目标数量小于等于0时, 记录的目标数量设为0,  切回初始化完成状态
        rpingTargetNum_ = 0;
        rpingState_ = RpingState::INITED;
    }

    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingGetTarget(u32 deviceId, u32 targetNum, RpingInput *input, int *targetStat)
{
    for (u32 i = 0; i < targetNum; i++) {
        //查询链路状态
        if (socketMaps_.find(std::string(input[i].dip.GetReadableIP())) == socketMaps_.end()) {
            HCCL_WARNING("[HCCN][HccnRpingGetTarget]Cannot get socket[%s]'s status.", input[i].dip.GetReadableIP());
            targetStat[i] = static_cast<int>(RpingLinkState::DISCONNECTED);
            continue;
        }
        HcclSocketStatus socketStatus = socketMaps_[std::string(input[i].dip.GetReadableIP())]->GetStatus();
    
        // 转换状态信息
        RpingLinkState linkStatus = ConvertHcclSocketStatus(socketStatus);
 
        // 记录查询结果
        targetStat[i] = static_cast<int>(linkStatus);
    }
 
    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingBatchPingStart(u32 deviceId, u32 pktNum, u32 interval, u32 timeout)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::RUN));
 
    // 调用hccp接口发起ping请求
    if (pingHandle_ == nullptr) {
        HCCL_ERROR("[HCCN][HccnRpingBatchPingStart]Device[%u] cannot start ping because it is not inited.", deviceId);
        return HCCL_E_NOT_FOUND;
    }
 
    // 计算内存空间能否保存全部的payload信息，内存不足的话不可以发起ping请求
    ping_buffer_info *bufferInfo = &(initInfo_.result);
    u32 targetNum = rpingTargetNum_;
    u32 payloadLen = pktNum * PING_TOTAL_PAYLOAD_MAX_SIZE * targetNum;
    if ((bufferInfo->buffer_size != 0) && (payloadLen >= bufferInfo->buffer_size)) {
        HCCL_ERROR("[HCCN][HccnRpingBatchPingStart]Buffer[%u] overflow threshold[%u].", payloadLen, bufferInfo->buffer_size);
        return HCCL_E_MEMORY;
    }

    ping_task_attr attr;
    attr.packet_cnt = pktNum;
    attr.packet_interval = interval;
    attr.timeout_interval = timeout;
 
    CHK_RET(hrtRaPingTaskStart(pingHandle_, &attr));
    HCCL_INFO("[HCCN][HccnRpingBatchPingStart]pingmesh task is started on device[%u].", deviceId);
    rpingState_ = RpingState::RUN;
 
    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingBatchPingStop(u32 deviceId)
{
    // 判断当前状态
    CHK_RET(RpingstateCheck(rpingState_, RpingState::STOP));
 
    // 调用hccp接口中止ping请求
    CHK_RET(hrtRaPingTaskStop(pingHandle_));
    HCCL_INFO("[HCCN][HccnRpingBatchPingStop]Device[%u] pingmesh task is manually stoped.", deviceId);
 
    rpingState_ = RpingState::STOP;
    return HCCL_SUCCESS;
}
 
HcclResult PingMesh::HccnRpingGetResult(u32 deviceId, u32 targetNum, RpingInput *input, RpingOutput *output)
{
    ping_target_result *resultInfo = new (std::nothrow) ping_target_result[targetNum];
    CHK_PRT_RET(resultInfo == nullptr, HCCL_ERROR("[HCCN][HccnRpingGetResult]Alloc result memory failed."),
        HCCL_E_MEMORY);
    RpingResultInfoInit(resultInfo, rdmaInfoMaps_, input, targetNum);
 
    // 调用hccp接口获取探测结果
    u32 num = targetNum; // resultinfo是一个带有返回值信息的数组, 对应的数组大小也需要返回，因此这里数组大小也传递指针
    HcclResult ret = hrtRaPingGetResults(pingHandle_, resultInfo, &num);
    if (ret == HCCL_E_AGAIN) {
        HCCL_WARNING("[HCCN][HccnRpingGetResult]Try again.");
        delete[] resultInfo;
        return HCCL_E_AGAIN;
    }
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[HCCN][HccnRpingGetResult]Device[%u] get result failed.", deviceId);
        delete[] resultInfo;
        return ret;
    }
    HCCL_INFO("[HCCN][HccnRpingGetResult]Device[%u] successfully gets [%u] results.", deviceId, num);
    
    GetResultFromReturnValue(resultInfo, output, targetNum);
 
    delete[] resultInfo;
    return HCCL_SUCCESS;
}

HcclResult PingMesh::HccnRpingRefillPayloadHead(u8 *originalHead, u32 payloadNum)
{
    for (u32 i = 0; i < payloadNum; i++) {
        RpingIpHead *ipHead = reinterpret_cast<RpingIpHead*>(originalHead);
        RpingIpHead ipHeadTmp;
        // 清零之前记录头信息
        errno_t memRet = memcpy_s(&ipHeadTmp, sizeof(RpingIpHead), ipHead, sizeof(RpingIpHead));
        CHK_PRT_RET(memRet != EOK,
            HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]copy head fail, ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
            memRet, &ipHeadTmp, sizeof(RpingIpHead), ipHead, sizeof(RpingIpHead)), HCCL_E_MEMORY);

        // 重填payload头
        RpingPayloadHead *head = reinterpret_cast<RpingPayloadHead*>(originalHead);   
        // 清零要重填的内存
        memRet = memset_s(originalHead, RPING_PAYLOAD_REFILL_LEN, 0, RPING_PAYLOAD_REFILL_LEN);
        CHK_PRT_RET(memRet != EOK,
            HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]clear first 136B fail, ret %d, destMaxSize %u, count %u",
            memRet, RPING_PAYLOAD_REFILL_LEN, RPING_PAYLOAD_REFILL_LEN), HCCL_E_MEMORY);
        // 清零payload头rsvd字段的内存
        memRet = memset_s(head->reserved, RPING_PAYLOAD_RSVD_LEN, 0, RPING_PAYLOAD_RSVD_LEN);
        CHK_PRT_RET(memRet != EOK, 
            HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]clear last 44B fail, ret %d, destMaxSize %u, count %u",
            memRet, RPING_PAYLOAD_RSVD_LEN, RPING_PAYLOAD_RSVD_LEN), HCCL_E_MEMORY);
        // 填充ip
        HcclInAddr srcIpBinary;
        HcclInAddr dstIpBinary;
        // 报文来自对端，因此srcIp和dstIp需要调换过来
        if (ipAddr_->GetFamily() == AF_INET) {
            srcIpBinary.addr.s_addr = ipHeadTmp.ipv4.dstIp;
            dstIpBinary.addr.s_addr = ipHeadTmp.ipv4.srcIp;
        } else {
            HcclInAddr *srcIpBinary6 = reinterpret_cast<HcclInAddr*>(ipHeadTmp.ipv6.srcIp);
            HcclInAddr *dstIpBinary6 = reinterpret_cast<HcclInAddr*>(ipHeadTmp.ipv6.dstIp);
            srcIpBinary = *dstIpBinary6;
            dstIpBinary = *srcIpBinary6;
        }
        HcclIpAddress srcIp = HcclIpAddress(ipAddr_->GetFamily(), srcIpBinary);
        u32 ipAddrStrLen = std::string(srcIp.GetReadableIP()).size();
        memRet = memcpy_s(head->srcIp, IP_ADDRESS_BUFFER_LEN, srcIp.GetReadableIP(), ipAddrStrLen);
        CHK_PRT_RET(memRet != EOK,
            HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead] Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
            memRet, head->srcIp, IP_ADDRESS_BUFFER_LEN, srcIp.GetReadableIP(), ipAddrStrLen), HCCL_E_MEMORY);
        HcclIpAddress dstIp = HcclIpAddress(ipAddr_->GetFamily(), dstIpBinary);
        ipAddrStrLen = std::string(dstIp.GetReadableIP()).size();
        memRet = memcpy_s(head->dstIp, IP_ADDRESS_BUFFER_LEN, dstIp.GetReadableIP(), ipAddrStrLen);
        CHK_PRT_RET(memRet != EOK,
            HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead] Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
            memRet, head->dstIp, IP_ADDRESS_BUFFER_LEN, dstIp.GetReadableIP(), ipAddrStrLen), HCCL_E_MEMORY);
        // 填充payloadLen
        if (payloadLenMap_.find(dstIp.GetReadableIP()) != payloadLenMap_.end()) {
            head->payloadLen = payloadLenMap_[dstIp.GetReadableIP()];
        }
        originalHead += BYTE_PER_TARGET_DEFAULT;
    }

    return HCCL_SUCCESS;
}

HcclResult PingMesh::HccnRpingGetPayload(u32 deviceId, void **payload, u32 *payloadLen)
{
    // 判断是否为payload配置过内存
    if (!isUsePayload_) {
        HCCL_DEBUG("[HCCN][HccnRpingGetPayload]not alloc memery on device[%u] for payload.", deviceId);
        *payload = nullptr;
        *payloadLen = 0;
        return HCCL_SUCCESS;
    }
    // 将payload信息从device拷贝到host
    ping_buffer_info *bufferInfo = &(initInfo_.result);
    CHK_PRT_RET(bufferInfo->buffer_size == 0,
        HCCL_ERROR("[HCCN][HccnRpingGetPayload]no memery on device[%u] for payload.", deviceId), HCCL_E_MEMORY);
    // payload_为空时，需要为其申请内存资源
    if (payload_ == nullptr) {
        payload_ = new (std::nothrow) u8[bufferInfo->buffer_size];
        CHK_PRT_RET(payload_ == nullptr,
            HCCL_ERROR("[HCCN][HccnRpingGetPayload]Get payload from device[%u] failed.", deviceId), HCCL_E_MEMORY);
    }
    // 从device拷贝内存
    HcclResult ret =
        hrtMemSyncCopyEx(payload_, bufferInfo->buffer_size, reinterpret_cast<void *>(bufferInfo->buffer_va),
        bufferInfo->buffer_size, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[HCCN][HccnRpingGetPayload]Get payload from device[%u] failed.", deviceId), ret);
    // 重填payload头
    u32 payloadNum = bufferInfo->buffer_size / BYTE_PER_TARGET_DEFAULT;
    u8 *payloadTmp = payload_;
    CHK_RET(HccnRpingRefillPayloadHead(payloadTmp, payloadNum));

    *payload = payload_;
    *payloadLen = bufferInfo->buffer_size;
    return HCCL_SUCCESS;
}

}